package com.ruoyi.generator.util;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.text.StrFormatter;
import cn.hutool.core.util.RandomUtil;
import com.ruoyi.generator.callback.MqttCallback;
import com.ruoyi.generator.domain.MqttProperty;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 自定义mqtt 客户端
 *
 * @author yuanqinglong
 */
@Data
@Slf4j
public class MQTTClient {

    private MqttClient client;

    private MqttProperty mqttProperty;

    private MqttConnectOptions setMqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(mqttProperty.getUserName());
        options.setPassword(mqttProperty.getPassWord().toCharArray());
        options.setConnectionTimeout(Math.toIntExact(mqttProperty.getTimeout()));
        options.setKeepAliveInterval(Math.toIntExact(mqttProperty.getKeepAlive()));
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        return options;
    }

    public static MQTTClient getMqttClient(MqttProperty mqttProperty) {
        MQTTClient mqttClient = new MQTTClient();
        mqttClient.setMqttProperty(mqttProperty);
        return mqttClient;
    }

    public void connect() throws MqttException {
        if (client == null) {
            client = new MqttClient(mqttProperty.getHost(), "java:client:"+RandomUtil.randomString(10)+ DateUtil.millisecond(DateUtil.date()),new MemoryPersistence());
            client.setCallback(new MqttCallback(this));
        }
        MqttConnectOptions options = setMqttConnectOptions();
        if (!client.isConnected()) {
            client.connect(options);
        } else {
            client.disconnect();
            client.connect(options);
        }
        log.debug(StrFormatter.format("MqttProperty ： {} connect success"));
    }
    /**
     * 发布，默认qos为0，非持久化
     *
     * @param pushMessage
     * @param topic
     */
    public void publish(String pushMessage, String topic) {
        publish(pushMessage, topic, 0, false);
    }
    /**
     * 发布消息
     *
     * @param pushMessage
     * @param topic
     * @param qos
     * @param retained:留存
     */
    public void publish(String pushMessage, String topic, int qos, boolean retained) {
        MqttMessage message = new MqttMessage();
        message.setPayload(pushMessage.getBytes());
        message.setQos(qos);
        message.setRetained(retained);
        MqttTopic mqttTopic = client.getTopic(topic);
        if (null == mqttTopic) {
            log.error("topic is not exist");
        }
        MqttDeliveryToken token;//Delivery:配送
        synchronized (this) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁，分析见文章最后补充
            try {
                token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
                token.waitForCompletion(1000L);
            } catch (MqttException e) {
                log.error(e.getMessage(),e);
            }
        }
    }
    /**
     * 订阅某个主题
     *
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            log.error(e.getMessage(),e);
        }
    }
    /**
     * 取消订阅主题
     *
     * @param topic 主题名称
     */
    public void cleanTopic(String topic) {
        if (client != null && client.isConnected()) {
            try {
                client.unsubscribe(topic);
            } catch (MqttException e) {
                log.error(e.getMessage(),e);
            }
        } else {
            log.error("取消订阅失败！");
        }
    }

}
